﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace TaskResult.TaskBlockingCollection
{
    /// <summary>
    /// 数据计算中心
    /// </summary>
    /// <typeparam name="T"></typeparam>
    public class DataProcessor<T>
    {
        private readonly int _threadCount;
        private readonly int _capacity;
        private readonly BlockingCollection<T> _collection;
        private readonly Action<T> _processAction;
        private readonly SemaphoreSlim _semaphore;

        public DataProcessor(Action<T> processAction, int threadCount = 5, int capacity = 20)
        {
            _threadCount = threadCount;
            _capacity = capacity;
            _processAction = processAction;
            _collection = new BlockingCollection<T>(_capacity);
            _semaphore = new SemaphoreSlim(threadCount);
        }
        public void Process(T[] largeDataset)
        {
            // 创建一个任务用于从数据集合读取数据并添加到阻塞集合
            Task producer = Task.Run(() =>
            {
                foreach (var data in largeDataset)
                {
                    // 如果集合已满，这个方法会阻塞
                    _collection.Add(data);
                }

                // 数据全部添加完后，标记集合不再添加新项
                _collection.CompleteAdding();
            });
            // 创建一个线程池，每个线程从阻塞集合中取数据进行处理
            Task[] consumers = new Task[_threadCount];
            for (int i = 0; i < _threadCount; i++)
            {
                consumers[i] = Task.Run(() =>
                {
                    // 如果集合为空，这个方法会阻塞，直到有新的项可用或者不再添加新项
                    foreach (var data in _collection.GetConsumingEnumerable())
                    {
                        // 使用传入的委托方法处理数据
                        _processAction(data);
                    }
                });
            }
            // 等待所有任务完成
            Task.WaitAll(consumers);
            producer.Wait();
        }
    }
}
